package org.example.importer;

import com.opencsv.CSVReader;
import com.opencsv.exceptions.CsvValidationException;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.example.hbase.HBaseUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.function.Consumer;

public class DataImporter {

    private static final Logger LOGGER = LoggerFactory.getLogger(DataImporter.class);

    public void importData(String resourceName) throws IOException, CsvValidationException {
        importData(resourceName, null);
    }

    public void importData(String resourceName, Consumer<Integer> progressCallback) throws IOException, CsvValidationException {
        LOGGER.info("准备从classpath下的 '{}' 文件导入数据...", resourceName);
        InputStream inputStream = getClass().getClassLoader().getResourceAsStream(resourceName);
        if (inputStream == null) {
            LOGGER.error("在classpath下找不到文件: '{}'。请确保文件存在于 'src/main/resources' 目录下。", resourceName);
            return;
        }

        try (
            InputStream rowCountStream = getClass().getClassLoader().getResourceAsStream(resourceName);
            CSVReader rowCounter = new CSVReader(new InputStreamReader(rowCountStream, StandardCharsets.UTF_8));
            InputStream dataStream = getClass().getClassLoader().getResourceAsStream(resourceName);
            CSVReader reader = new CSVReader(new InputStreamReader(dataStream, StandardCharsets.UTF_8));
            Table table = HBaseUtil.getConnection().getTable(HBaseUtil.TABLE_NAME)
        ) {
            rowCounter.readNext();
            int totalRows = 0;
            while (rowCounter.readNext() != null) {
                totalRows++;
            }
            LOGGER.info("CSV文件总行数（不含标题）: {}", totalRows);

            String[] headers = reader.readNext(); // 读取标题行
            if (headers == null) {
                LOGGER.warn("CSV文件 '{}' 为空或缺少标题行。", resourceName);
                return;
            }

            LOGGER.info("成功读取标题行，开始逐行导入数据...");
            String[] line;
            int lineCount = 0;
            while ((line = reader.readNext()) != null) {
                lineCount++;
                if (line.length != headers.length) {
                    LOGGER.warn("第 {} 行格式错误，已跳过: {}", lineCount, String.join(",", line));
                    continue;
                }

                String recordId = getColumnValue("Record ID", headers, line);
                if(recordId == null || recordId.isEmpty()){
                    // Record ID是行键，不能为空
                    LOGGER.warn("第 {} 行的Record ID为空，已跳过。", lineCount);
                    continue;
                }

                Put put = new Put(Bytes.toBytes(recordId));

                // 根据imagee2中的定义进行映射
                // info 列族
                put.addColumn(HBaseUtil.CF_INFO, Bytes.toBytes("NHTSA Campaign"), Bytes.toBytes(getColumnValue("NHTSA Campaign", headers, line)));
                put.addColumn(HBaseUtil.CF_INFO, Bytes.toBytes("Manufacturer Campaign"), Bytes.toBytes(getColumnValue("Manufacturer Campaign", headers, line)));
                put.addColumn(HBaseUtil.CF_INFO, Bytes.toBytes("Recall Initiative"), Bytes.toBytes(getColumnValue("Recall Initiative", headers, line)));

                // vehicle 列族
                put.addColumn(HBaseUtil.CF_VEHICLE, Bytes.toBytes("Vehicle Make"), Bytes.toBytes(getColumnValue("Vehicle Make", headers, line)));
                put.addColumn(HBaseUtil.CF_VEHICLE, Bytes.toBytes("Vehicle Model"), Bytes.toBytes(getColumnValue("Vehicle Model", headers, line)));
                put.addColumn(HBaseUtil.CF_VEHICLE, Bytes.toBytes("Model Year"), Bytes.toBytes(getColumnValue("Model Year", headers, line)));
                put.addColumn(HBaseUtil.CF_VEHICLE, Bytes.toBytes("Vehicle Manufacturer"), Bytes.toBytes(getColumnValue("Vehicle Manufacturer", headers, line)));

                // recall 列族
                put.addColumn(HBaseUtil.CF_RECALL, Bytes.toBytes("Recall Type"), Bytes.toBytes(getColumnValue("Recall Type", headers, line)));
                put.addColumn(HBaseUtil.CF_RECALL, Bytes.toBytes("Recall Component"), Bytes.toBytes(getColumnValue("Recall Component", headers, line)));
                put.addColumn(HBaseUtil.CF_RECALL, Bytes.toBytes("Manufacture Start Date"), Bytes.toBytes(getColumnValue("Manufacture Start Date", headers, line)));
                put.addColumn(HBaseUtil.CF_RECALL, Bytes.toBytes("Manufacture End Date"), Bytes.toBytes(getColumnValue("Manufacture End Date", headers, line)));
                put.addColumn(HBaseUtil.CF_RECALL, Bytes.toBytes("Estimated Units"), Bytes.toBytes(getColumnValue("Estimated Units", headers, line)));
                put.addColumn(HBaseUtil.CF_RECALL, Bytes.toBytes("Recall Manufacturer"), Bytes.toBytes(getColumnValue("Recall Manufacturer", headers, line)));
                put.addColumn(HBaseUtil.CF_RECALL, Bytes.toBytes("Recall Notification Date"), Bytes.toBytes(getColumnValue("Recall Notification Date", headers, line)));

                table.put(put);

                // 向前端UI报告进度百分比
                if (progressCallback != null && totalRows > 0) {
                    if (lineCount % 500 == 0 || lineCount == totalRows) { // 每500行或最后一行时报告进度
                        int percentage = (int) (((double) lineCount / totalRows) * 100);
                        progressCallback.accept(percentage);
                    }
                }
                
                // 在后台控制台打印导入的行数
                if (lineCount % 1000 == 0) {
                    LOGGER.info("已成功导入 {} 行数据...", lineCount);
                }
            }
            LOGGER.info("数据导入完成。总共处理了 {} 行。", lineCount);
        }
    }

    private String getColumnValue(String columnName, String[] headers, String[] line) {
        for (int i = 0; i < headers.length; i++) {
            if (headers[i].equalsIgnoreCase(columnName)) {
                return line[i];
            }
        }
        return ""; // 如果找不到列，返回空字符串
    }
} 